package com.gitee.broadcast
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
/*
  将Configuration对象作为参数传递
 */
object WithParametersTest {
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val toFilter = env.fromElements(1, 2, 3)

    val c = new Configuration()
    c.setInteger("limit", 1)

    val result: DataSet[Int] = toFilter.filter(new RichFilterFunction[Int]() {
      var limit = 0

      override def open(config: Configuration): Unit = {
        limit = config.getInteger("limit", 0)
      }

      def filter(in: Int): Boolean = {
        in > limit
      }
    }).withParameters(c)
    result.print()

  }

}
